package com.atbeijing.kafka.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

public class Consumer {
    public static void main(String[] args) {
            Properties conf = new Properties();
            try {
                conf.load(Consumer.class.getClassLoader().getResourceAsStream("consumer.properties"));
            } catch (IOException e) {
                e.printStackTrace();
            }

            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(conf);
            //消费指定topic
            consumer.subscribe(Collections.singleton("first"));
            //循环拉取
//            while (true) {
                ConsumerRecords<String, String> poll = consumer.poll(Duration.ofSeconds(10));
                //消费消息
                for (ConsumerRecord<String, String> consumerRecord : poll) {
                    System.out.println(consumerRecord);
                }
//            }

            //关闭consumer
//            consumer.close();
    }
}
